package analysis.ProfessionalSchoolPlan;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ProfessionalSchoolPlanMain {
    public static void main(String[] args) throws Exception {
        //创建Configuration对象时，会自动加载默认配置参数
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);

        // 1、封装参数：main方法所在的class
        job.setJarByClass(ProfessionalSchoolPlanMain.class);

        // 2、封装参数： 本次job所要调用的Mapper实现类、Reducer实现类
        job.setMapperClass(ProfessionalSchoolPlanMapper.class);
        job.setReducerClass(ProfessionalSchoolPlanReducer.class);

        // 3、封装参数：本次job的Mapper实现类、Reducer实现类产生的结果数据的key、value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 4、封装参数：本次job要处理的输入数据集所在路径、最终结果的输出路径
        Path inpath = new Path("F:\\codes\\hadoop\\schoolplan\\src\\main\\resources\\preprocess\\preprocessedData.csv");
        Path outPath = new Path("F:\\codes\\hadoop\\schoolplan\\src\\main\\resources\\analysis\\ProfessionalSchoolPlan");
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(outPath)) {
            System.out.println("输出路径 " + outPath.getName() + " 已存在,默认删除");
            fs.delete(outPath, true);
        }
        FileInputFormat.setInputPaths(job, inpath);
        FileOutputFormat.setOutputPath(job, outPath);

        // 5、封装参数：想要启动的reduce task的数量
        job.setNumReduceTasks(1);

        // 6、提交job给yarn
        boolean res = job.waitForCompletion(true);

        System.exit(res ? 0 : -1);
    }
}
